package com.xxxx.spark.operator

import com.xxxx.spark.WorldCount
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @program: day0305
 * @description: 尽信书，则不如无书
 * @author: CoreDao
 * @create: 2021-03-05 10:53
 * */
/**
 * 持久化算子cache，对RDD进行持久化到内存的操作（中间结果的持久化），能提升性能，有一定局限性（内存大小）
 * job是串行的
 * job里面是并行
 */
object ControlFun {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local").setAppName("cache")
    val sc = new SparkContext(conf)
    //checkpoint使用需要设置数据存储路径
    sc.setCheckpointDir("./checkpoint")
    val file = WorldCount.getClass.getClassLoader.getResource("data/NASA_access_log_Aug95").getFile

    //cache() = persist() 只放在内存
    var lineRDD = sc.textFile(file)

    //lineRDD = lineRDD.cache()
    //数据量级决定用哪个等级
//    lineRDD = lineRDD.persist(StorageLevel.DISK_ONLY)

    lineRDD.checkpoint()

    //此RDD共计1569898 条数据，初始化数据以及cache用时1735
    //cache后
    //此RDD共计1569898 条数据，初始化数据以及cache用时2520
    //DISK_ONLY后
    //此RDD共计1569898 条数据，初始化数据以及cache用时3951
    val startTime = System.currentTimeMillis()
    val count = lineRDD.count()
    val endTime = System.currentTimeMillis()
    println(s"此RDD共计$count 条数据，初始化数据以及cache用时${endTime-startTime}")

    //此RDD共计1569898 条数据，初始化数据以及cache用时746
    //此RDD共计1569898 条数据，初始化数据以及cache用时118
    //此RDD共计1569898 条数据，初始化数据以及cache用时607
    val startTime2 = System.currentTimeMillis()
    val count2 = lineRDD.count()
    val endTime2 = System.currentTimeMillis()
    println(s"此RDD共计$count2 条数据，初始化数据以及cache用时${endTime2-startTime2}")

  }
}
